package org.bdware.repository.handler;

import com.google.gson.Gson;
import com.google.gson.reflect.TypeToken;
import org.apache.log4j.Logger;
import org.bdware.doip.core.doipMessage.DoipMessage;
import org.bdware.doip.core.doipMessage.DoipResponseCode;
import org.bdware.doip.core.model.digitalObject.*;
import org.bdware.doip.core.model.metadata.MetaDO;
import org.bdware.doip.core.model.metadata.SearchParameter;
import org.bdware.doip.core.model.metadata.SearchResult;
import org.bdware.doip.core.utils.DoipGson;
import org.bdware.doip.endpoint.doipClient.DoipClientImpl;
import org.bdware.doip.endpoint.doipClient.DoipMessageCallback;
import org.bdware.doip.endpoint.doipServer.DoipServiceInfo;
import org.bdware.doip.endpoint.doipServer.RegistryHandlerBase;
import org.bdware.doip.implementation.DoIndexer;
import org.bdware.irp.client.GlobalIrpClient;
import org.bdware.irp.client.IrpClient;
import org.bdware.irp.exception.IrpClientException;
import org.bdware.irp.stateinfo.DoStateInfo;
import org.bdware.repository.storage.BDRepoStorage;
import org.bdware.repository.storage.RemoteReg;
import org.bdware.repository.utils.Utils;
import org.rocksdb.RocksDBException;

import java.io.IOException;
import java.net.URI;
import java.util.HashMap;
import java.util.List;

public class BDRepositoryHandler extends RegistryHandlerBase {
    static Logger logger = Logger.getLogger(BDRepositoryHandler.class);
    BDRepoStorage store;
    DoIndexer indexer;
    HashMap<String, RemoteReg> remoteRegs;
    static final String REMOTE_REG = "REMOTEREG";
    static final String DO_OWNER = "doOwner";

    public BDRepositoryHandler(DoipServiceInfo info, BDRepoStorage store) throws RocksDBException {
        super(info);
        this.store = store;
        if (store.getRaw(REMOTE_REG.getBytes()) != null) {
            remoteRegs = loadRegList();
        } else
            remoteRegs = new HashMap<>();
    }

    public BDRepositoryHandler(DoipServiceInfo info, BDRepoStorage store, DoIndexer indexer) throws RocksDBException {
        this(info, store);
        this.indexer = indexer;
    }

    @Override
    public DoipMessage handleHello(DoipMessage request) {
        return replyDoipServiceInfo(request);
    }

    @Override
    public DoipMessage handleListOps(DoipMessage request) {
        return replyAllOperations(request);
    }

    @Override
    public DoipMessage handleCreate(DoipMessage request) {
        try {
            if (!request.header.parameters.id.equals(serviceInfo.id)) {
                return replyStringWithStatus(request, "Wrong operation targetID, should be: " + serviceInfo.id, DoipResponseCode.Invalid);
            }
            String doOwner = serviceInfo.owner;
            if (request.credential != null) doOwner = request.credential.getSigner();
            DigitalObject digitalObject = request.body.getDataAsDigitalObject();
            if (digitalObject.id != null && !digitalObject.id.equals("")) {
                DigitalObject ret = store.getDoByID(digitalObject.id);
                if (ret != null) {
                    return replyStringWithStatus(request, "Failed: the data of this DO is already existed.", DoipResponseCode.DoAlreadyExist);
                }
            }
            if (digitalObject.type == DoType.DOList) {
                DOList doList = DOList.fromDigitalObject(digitalObject);
                List<DigitalObject> dos = doList.getDOList();
                IrpClient irpClient = GlobalIrpClient.getGlobalClient();
                DoStateInfo doHR = new DoStateInfo(doOwner, serviceInfo.id);
                if (digitalObject.id == null || digitalObject.id.equals("")) {   //register and set ID for all DOs
                    List<String> ids = irpClient.batchRegister(doHR, 1 + dos.size());
                    doList = new DOList(ids.get(0));
                    for (int i = 0; i < dos.size(); i++) {      //set each DO ID
                        dos.get(i).id = ids.get(i + 1);
                        doList.addDO(dos.get(i));
                    }
                }
                for (DigitalObject aDo : dos) {      //store each DO Element in DOList
                    aDo.addAttribute(DO_OWNER, doOwner);
                    store.setRequester(doOwner).storeDo(aDo);
                    indexDO(aDo);
                }
                //clean the elements data,and store DOList
                for (Element e : doList.elements) {
                    e.excludeData();
                }
                doList.addAttribute(DO_OWNER, doOwner);
                store.setRequester(doOwner).storeDo(doList);
                indexDO(doList);
                return replyDO(request, doList);
            } else {
                if (digitalObject.id == null || digitalObject.id.equals("")) {        //need to register a new DOID
                    logger.debug("register DO for user");
                    IrpClient irpClient = GlobalIrpClient.getGlobalClient();
                    DoStateInfo doHR = new DoStateInfo(doOwner, serviceInfo.id);
                    digitalObject.id = irpClient.register(doHR);
                } else {
                    DoStateInfo doHR = new DoStateInfo(doOwner, serviceInfo.id);
                    doHR.identifier = digitalObject.id;
                    new Thread(() -> {
                        try {
                            GlobalIrpClient.getGlobalClient().reRegister(doHR);
                        } catch (IrpClientException e) {
                            e.printStackTrace();
                        }
                    }).start();
                }
                logger.debug("Message body length: " + request.body.getLength());
                logger.info("create DO identifier : " + digitalObject.id);
                digitalObject.addAttribute(DO_OWNER, doOwner);

                if (digitalObject.elements != null) {
                    for (Element e : digitalObject.elements) {
                        //对于文件类型的element需要特殊处理
                        if ("file".equals(e.type)) {
                            //TODO 对于文件类型的element，属性中应该明确下文件类型,content-type
                            if (e.attributes != null && e.attributes.get("file") != null) {
                                String downloadUrl = e.attributes.get("file").getAsString();
                                byte[] data = Utils.getFileFromRemote(downloadUrl);
                                e.setData(data);
                            }
                        }
                    }
                }

                store.setRequester(doOwner).storeDo(digitalObject);
                indexDO(digitalObject);
                if (digitalObject.elements != null)
                    for (Element e : digitalObject.elements) {
                        e.excludeData();
                    }
                return replyDO(request, digitalObject);
            }
        } catch (Exception e) {
            e.printStackTrace();
            return replyStringWithStatus(request, "Server Exception: " + e.getMessage(), DoipResponseCode.MoreThanOneErrors);
        }
    }

    @Override
    public DoipMessage handleUpdate(DoipMessage request) {
        try {
            String doOwner = serviceInfo.owner;
            if (request.credential != null) doOwner = request.credential.getSigner();
            DigitalObject digitalObject = request.body.getDataAsDigitalObject();
            digitalObject.addAttribute(DO_OWNER, doOwner);
            if (!request.header.parameters.id.equals(digitalObject.id)) {
                return replyStringWithStatus(request, "Invalid request: different identifiers in DO and message header.", DoipResponseCode.Invalid);
            }
            DigitalObject ret = store.getDoByID(digitalObject.id);
            if (ret == null) {
                return replyStringWithStatus(request, "Failed: DO is not existed.", DoipResponseCode.DoNotFound);
            }
            if (checkUpdateAuth(digitalObject))
                return replyStringWithStatus(request, "Failed: unAuthOperation", DoipResponseCode.UnAuth_Op);
            store.setRequester(doOwner).updateDo(digitalObject);
            if (digitalObject.attributes != null && digitalObject.attributes.get("metadata") != null) { //update index
                deleteIndex(digitalObject.id);
                indexDO(digitalObject);
            }
            if (digitalObject.elements != null)
                for (Element e : digitalObject.elements) {
                    e.excludeData();
                }
            return replyDO(request, digitalObject);
        } catch (Exception e) {
            e.printStackTrace();
            return replyStringWithStatus(request, "Server Exception: " + e.getMessage(), DoipResponseCode.MoreThanOneErrors);
        }
    }

    private boolean checkUpdateAuth(DigitalObject newDO) {
        DigitalObject oldDO = store.getDoByID(newDO.id);
        if (getDoOwner(oldDO) == null) return true;
        return getDoOwner(oldDO).equals(getDoOwner(newDO));
    }

    String getDoOwner(DigitalObject digitalObject) {
        return digitalObject.attributes.get(DO_OWNER) == null ? null : digitalObject.attributes.get(DO_OWNER).getAsString();
    }

    @Override
    public DoipMessage handleDelete(DoipMessage request) {
        try {
            DigitalObject ret = store.getDoByID(request.header.parameters.id);
            if (ret == null) {
                return replyStringWithStatus(request, "Failed: DO is not existed.", DoipResponseCode.DoNotFound);
            }
            String requester = null;
            if (request.credential != null) requester = request.credential.getSigner();
            store.setRequester(requester).deleteDoByID(request.header.parameters.id);
            deleteIndex(ret.id);
            IrpClient irpClient = GlobalIrpClient.getGlobalClient();
            irpClient.unRegister(ret.id);
            return replyNull(request);
        } catch (Exception e) {
            e.printStackTrace();
            return replyStringWithStatus(request, "Server Exception: " + e.getMessage(), DoipResponseCode.MoreThanOneErrors);
        }
    }

    @Override
    public DoipMessage handleRetrieve(DoipMessage request) {
        try {
            DigitalObject ret = store.getDoByID(request.header.parameters.id);
            if (ret == null) {
                return replyStringWithStatus(request, "Failed: DO is not existed.", DoipResponseCode.DoNotFound);
            }
            if (ret.elements != null) {
                //如果是获取element
                if (request.header.parameters.attributes != null && request.header.parameters.attributes.has("element")) {
                    for (Element e : ret.elements) {
                        if (e.id.equals(request.header.parameters.attributes.get("element").getAsString())) {
                            return replyString(request, e.getData() == null ? "" : new String(e.getData()));
                        }
                    }
                    return replyStringWithStatus(request,
                            "element not found: " + request.header.parameters.attributes.get("element"),
                            DoipResponseCode.DoNotFound);
                }
                //不需要获取Elementdata
                if (request.header.parameters.attributes == null ||
                        !request.header.parameters.attributes.has("includeElementData")) {
                    logger.debug("clean element data");
                    for (Element e : ret.elements) {
                        e.excludeData();
                    }
                }
            }
            //对于特殊的API类型的DO，不是直接返回DigitalObject，而是需要请求API数据，返回真正的结果
            if (ret.elements != null) {
                DoWithAPI doWithAPI = new DoWithAPI(ret);
                doWithAPI.retrieveApiData(null);
                ret = doWithAPI;
            }
            return replyDO(request, ret);
        } catch (Exception e) {
            e.printStackTrace();
            return replyStringWithStatus(request, "Server Exception: " + e.getMessage(), DoipResponseCode.MoreThanOneErrors);
        }
    }

    @Override
    public DoipMessage handleSearch(DoipMessage request) {
        if (indexer == null) return replyStringWithStatus(request, "Unsupported Operation!", DoipResponseCode.Declined);
        if (request.header.parameters.attributes == null) {
            logger.info("invalid search request: attributes not found");
            return replyStringWithStatus(request, "invalid search request: attributes not found.", DoipResponseCode.Invalid);
        }
        SearchParameter sp = new SearchParameter(request.header.parameters.attributes.get("query").getAsString(),
                !request.header.parameters.attributes.has("pageNum") ? 0 : request.header.parameters.attributes.get("pageNum").getAsInt(),
                !request.header.parameters.attributes.has("pageSize") ? 0 : request.header.parameters.attributes.get("pageSize").getAsInt(),
                !request.header.parameters.attributes.has("sortFields") ? "" : request.header.parameters.attributes.get("sortFields").getAsString(),
                !request.header.parameters.attributes.has("type") ? null : request.header.parameters.attributes.get("type").getAsString());
        try {
            SearchResult sr = indexer.search(sp);
            return replyString(request, DoipGson.getDoipGson().toJson(sr));
        } catch (IOException e) {
            e.printStackTrace();
            return replyStringWithStatus(request, "Server Exception: " + e.getMessage(), DoipResponseCode.MoreThanOneErrors);
        }
    }

    HashMap<String, RemoteReg> loadRegList() {
        try {
            byte[] regBytes = store.getRaw(REMOTE_REG.getBytes());
            return new Gson().fromJson(new String(regBytes), new TypeToken<HashMap<String, RemoteReg>>() {
            }.getType());
        } catch (RocksDBException e) {
            e.printStackTrace();
        }
        return null;
    }

    void saveRegList() {
        try {
            store.putRaw(REMOTE_REG.getBytes(), new Gson().toJson(remoteRegs).getBytes());
        } catch (RocksDBException e) {
            e.printStackTrace();
        }
    }

    private void indexDO(DigitalObject digitalObject) throws IOException {
        if (indexer == null) return;
        if (digitalObject.attributes == null || digitalObject.attributes.get("metadata") == null) return;
        indexer.indexDigitalObject(MetaDO.fromDO(digitalObject));
        distributedIndex(digitalObject);
    }

    private void deleteIndex(String doID) {
        if (indexer == null) return;
        indexer.deleteDigitalObject(doID);
    }

    public void addLevel2RemoteRegistry(String regID, String url) {
        RemoteReg newReg = new RemoteReg(regID, url, 2);
        remoteRegs.put(regID, newReg);
        saveRegList();
    }

    public void addRemoteRegistry(RemoteReg newReg) {
        remoteRegs.put(newReg.id, newReg);
        saveRegList();
    }

    public void removeRemoteRegs(String regID) {
        remoteRegs.remove(regID);
        saveRegList();
    }

    public HashMap<String, RemoteReg> getRemoteRegs() {
        return remoteRegs;
    }

    void distributedIndex(DigitalObject digitalObject) {
        logger.info("remote regs:\n" + new Gson().toJson(remoteRegs));
        if (remoteRegs == null) return;
        AggregateFlag aggregateFlag = new AggregateFlag(remoteRegs.size());
        if (aggregateFlag.isFinished()) return;
        MetaDO metaDO = MetaDO.fromDO(digitalObject);
        assert metaDO != null;
        metaDO.addAttribute(DO_OWNER, getDoOwner(digitalObject));
        IndexCallback cb = new IndexCallback(aggregateFlag);
        for (RemoteReg reg : remoteRegs.values()) {
            if (reg.indexLevel != 2) continue;
            logger.debug("try to create index in: " + reg.id);
            new Thread(() -> {
                DoipClientImpl client = new DoipClientImpl();
                client.connect(reg.url);
                logger.debug("send search message to :" + reg.url);
                client.create(reg.id, metaDO, cb);
            }).start();
        }
    }

    static class IndexCallback implements DoipMessageCallback {
        AggregateFlag wf;

        public IndexCallback(AggregateFlag flag) {
            wf = flag;
        }

        @Override
        public void onResult(DoipMessage msg) {
            logger.debug("receive a search resp: " + msg.body.getDataAsJsonString());
            if (wf.isFinished()) return;
            if (msg.header.parameters.response != DoipResponseCode.Success) {
                logger.warn("index DO failed, target: " + msg.header.parameters.id + "; error msg: " + msg.body.getDataAsJsonString());
                return;
            }
            logger.info("Success index DO:" + msg.body.getDataAsJsonString());
            synchronized (wf) {
                wf.receive();
            }
        }
    }

    static class AggregateFlag {
        int received = 0;
        final int total;

        public AggregateFlag(int total) {
            this.total = total;
        }

        public void receive() {
            received++;
        }

        public boolean isFinished() {
            return received >= total;
        }

        void abort() {
            received = total;
        }
    }
}
